package com.binance.notification.web.service.impl;

import java.io.IOException;
import java.io.InputStream;
import java.math.BigDecimal;
import java.math.MathContext;
import java.math.RoundingMode;
import java.util.Arrays;
import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;

import com.github.rholder.retry.Attempt;
import com.github.rholder.retry.RetryException;
import com.github.rholder.retry.RetryListener;
import com.github.rholder.retry.Retryer;
import com.github.rholder.retry.RetryerBuilder;
import com.github.rholder.retry.StopStrategies;
import com.github.rholder.retry.WaitStrategies;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.core.env.Environment;
import org.springframework.dao.DataAccessException;
import org.springframework.dao.InvalidDataAccessApiUsageException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.ReturnType;
import org.springframework.data.redis.core.DefaultTypedTuple;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.SessionCallback;
import org.springframework.data.redis.core.ZSetOperations.TypedTuple;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.FileCopyUtils;
import org.springframework.util.StopWatch;
import org.springframework.web.client.RestTemplate;

import com.alibaba.fastjson.JSON;
import com.binance.assetservice.api.IProductApi;
import com.binance.assetservice.vo.ProductItemVO;
import com.binance.assetservice.vo.request.product.GetProductItemRequest;
import com.binance.assetservice.vo.response.product.ProductItemResponse;
import com.binance.master.annotations.DefaultDB;
import com.binance.master.error.BusinessException;
import com.binance.master.error.GeneralCode;
import com.binance.master.models.APIRequest;
import com.binance.master.models.APIResponse;
import com.binance.master.utils.CopyBeanUtils;
import com.binance.master.utils.DateUtils;
import com.binance.notification.api.push.PushCommand;
import com.binance.notification.api.push.notification.AndroidPayload;
import com.binance.notification.api.push.notification.AppPayload;
import com.binance.notification.api.push.notification.IosPayload;
import com.binance.notification.api.push.notification.MultipleNotificationPacket;
import com.binance.notification.api.push.notification.PcPayload;
import com.binance.notification.api.push.notification.WebPayload;
import com.binance.notification.data.mapper.AlertPriceMapper;
import com.binance.notification.data.models.AlertPrice;
import com.binance.notification.web.kafka.Productor;
import com.binance.notification.web.push.PushResult;
import com.binance.notification.web.push.RequestCallback;
import com.binance.notification.web.service.IAlertPriceService;
import com.binance.notification.web.service.IPushService;
import com.binance.notification.web.service.ws.ProductRealtimePriceService;
import com.binance.notification.web.utils.Worker;
import com.google.common.base.Objects;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;

import cn.jiguang.common.ServiceHelper;
import lombok.extern.slf4j.Slf4j;
import redis.clients.jedis.JedisCluster;

@Component
@Slf4j
public class AlertPriceServiceImpl implements IAlertPriceService {

    private static final String LASTEST_SYMBOL_PRICE_KEY_PREV = "lastest_symbol_price";
    private static final LinkedBlockingQueue<String> ALERT_PRICE_STRING_QUEUE = new LinkedBlockingQueue<>();
    private static final LinkedBlockingQueue<String> ALERT_PRICE_PAIR_QUEUE = new LinkedBlockingQueue<>();
    private static final LinkedBlockingQueue<String> ALERT_PRICE_SUCCESS_QUEUE = new LinkedBlockingQueue<>();
    private static final LinkedBlockingQueue<String> ALERT_PRICE_FAILURE_QUEUE = new LinkedBlockingQueue<>();
    private static final String CACHE_KEY_PREFIX = "PA:";
    @Resource
    private RestTemplate restTemplate;
    @Resource
    private AlertPriceMapper alertPriceMapper;
    @Resource
    private RedisTemplate<String, Object> redisTemplate;
    @Resource
    private Productor productor;
    @Value("${jpush.alert.price.enabled:false}")
    private boolean alertPriceEnabled;
    @Value("${jpush.alert.price.android.enabled:false}")
    private boolean alertPriceAndroidEnabled;
    @Value("${jpush.alert.price.ios.enabled:false}")
    private boolean alertPriceIosEnabled;
    @Value("${jpush.alert.price.web.enabled:false}")
    private boolean alertPriceWebEnabled;
    @Value("${jpush.alert.price.pc.enabled:false}")
    private boolean alertPricePcEnabled;
    // 每次http请求推送时最大批量数，小于等于0表示不采用批量方式
    @Value("${jpush.alert.price.batch.count:1000}")
    private int batchCount;
    // 是否将待推送信息发送到kafka中，从而将压力分散到各个节点
    @Value("${jpush.alert.price.kafka:false}")
    private boolean alertByKafka;
    @Value("${jpush.alert.price.minnumStrategy:0}")
    private int minnumStrategy;
    @Value("${jpush.alert.price.maxnumStrategy:0}")
    private int maxnumStrategy;
    @Autowired
    private Environment env;
    @Autowired
    private IPushService pushService;
    @Resource
    private IProductApi productApi;
    @Autowired
    private ProductRealtimePriceService productRealtimePriceService;
    @Autowired
    private Worker worker;
    private String luaScript;

    private final Retryer<ProductItemVO> retryer = RetryerBuilder.<ProductItemVO>newBuilder()
            .retryIfExceptionOfType(Throwable.class).withRetryListener(new RetryListener() {

                @Override
                public <V> void onRetry(Attempt<V> attempt) {
                    if (attempt.hasException()) {
                        System.out.println(System.currentTimeMillis());
                        log.warn(String.format("%d retry after %d ms", attempt.getAttemptNumber(),
                                attempt.getDelaySinceFirstAttempt()), attempt.getExceptionCause());
                    }
                }

            }).withWaitStrategy(WaitStrategies.fixedWait(5, TimeUnit.SECONDS))
            .withStopStrategy(StopStrategies.stopAfterAttempt(5)).build();

    /**
     * 只将小数部分格式化位8位小数（不足后面补零）
     *
     * @param value
     * @return
     */
    private static String decimal_8(Object value) {
        // DecimalFormat df = new DecimalFormat("0.00000000");
        // return df.format(value);
        return value.toString();
    }

    /**
     * 将整数部分格式化位10位（不足10位前面补零），小数部分格式化位8位（不足后面补零）
     *
     * @param value
     * @return
     */
    private static String decimal_10_8(Object value) {
        // DecimalFormat df = new DecimalFormat("0000000000.00000000");
        // return df.format(value);
        return value.toString();
    }

    @PostConstruct
    public void init() {
        byte[] chars = new byte[0];
        try (InputStream in = this.getClass().getResourceAsStream("filterAlertPrice.lua");) {
            chars = FileCopyUtils.copyToByteArray(in);
        } catch (IOException e) {
            log.error("加载redis脚本出错", e);
            System.exit(-1);
        }
        luaScript = new String(chars);
        for (int i = 0; i < 5; i++) {
            worker.submit(() -> {
                pushFromAlertPriceQueue();
            }, () -> {
                pushFromPairQueue();
            }, () -> {
                success();
            }, () -> {
                failure();
            });
        }
    }

    @Override
    @Transactional(transactionManager = DefaultDB.TRANSACTION)
    public AlertPrice insert(AlertPrice price) throws Exception {
        log.info("插入数据:{}", price);
        BigDecimal newPrice = price.getLastPrice();
        BigDecimal setPrice = price.getPrice();
        int a = setPrice.compareTo(newPrice);
        MathContext mc = new MathContext(8, RoundingMode.HALF_UP);
        if (a > 0) {
            price.setSide((byte) 1);
            if (setPrice.divide(newPrice, mc).compareTo(BigDecimal.valueOf(100)) > 0) {
                throw new BusinessException(GeneralCode.AP_BREAK_SIDE_UP_PRICE_MUST_OVER_NOW);
            }
        } else if (a < 0) {
            price.setSide((byte) 2);
            if (newPrice.divide(setPrice, mc).compareTo(BigDecimal.valueOf(100)) > 0) {
                throw new BusinessException(GeneralCode.AP_BREAK_SIDE_DOWN_PRICE_MUST_UNDER_NOW);
            }
        } else {
            throw new BusinessException(GeneralCode.AP_BREAK_MUST_NOT_NE_NOW_PRICE);
        }
        ProductItemVO productItemVO = this.checkProduct(price.getBaseAsset(), price.getQuoteAsset());
        price.setCreateTime(DateUtils.getNewUTCDate());
        this.beforeSave(price, productItemVO);
        this.alertPriceMapper.insert(price);
        this.cacheAlertPrice(true, true, price);
        return price;
    }

    private ProductItemVO checkProduct(String baseAsset, String quoteAsset) throws ExecutionException, RetryException {
        GetProductItemRequest body = new GetProductItemRequest();
        body.setSymbol(baseAsset + quoteAsset);
        APIRequest<GetProductItemRequest> request = APIRequest.instance(body);

        ProductItemVO productItemVO = retryer.call(() -> {
            APIResponse<ProductItemResponse> response = productApi.getProductItem(request);
            if (response == null) {
                throw new RuntimeException("请求异常");
            }
            return (response.getData() == null || CollectionUtils.isEmpty(response.getData().getProductItems())) ? null
                    : response.getData().getProductItems().get(0);
        });

        if (productItemVO == null) {
            throw new BusinessException(GeneralCode.PRODUCT_NOT_EXIST);
        }
        return productItemVO;
    }

    @Override
    @Transactional(transactionManager = DefaultDB.TRANSACTION)
    public int update(AlertPrice price) throws Exception {
        log.info("更新数据:{}", price);
        int count = this.alertPriceMapper.updateByPrimaryKeySelective(price);
        if (count > 0) {
            this.cacheAlertPrice(true, true, price);
        }
        return count;
    }

    @Override
    @Transactional(transactionManager = DefaultDB.TRANSACTION)
    public int remove(List<Long> idList, String userId) throws Exception {
        log.info("根据条件'{}，{}'删除数据...", idList, userId);
        List<AlertPrice> list = this.alertPriceMapper.selectByPrimaryKey(idList, userId);
        int count = 0;
        if (CollectionUtils.isEmpty(idList)) {
            count = this.alertPriceMapper.deleteByPrimaryKey(null, userId);
        } else {
            count = this.alertPriceMapper.deleteByPrimaryKeyAndUserId(idList, Lists.newArrayList(userId));
        }
        if (count > 0) {
            this.redisTemplate.execute(new SessionCallback<Void>() {

                @SuppressWarnings({"rawtypes", "unchecked"})
                @Override
                public Void execute(RedisOperations operations) throws DataAccessException {
                    for (AlertPrice ap : list) {
                        operations.opsForZSet().remove(ap.getBaseAsset() + ap.getQuoteAsset() + ap.getSide(),
                                serializer(ap));
                    }
                    if (CollectionUtils.isEmpty(idList)) {
                        operations.delete(CACHE_KEY_PREFIX + userId);
                    } else {
                        operations.opsForHash().delete(CACHE_KEY_PREFIX + userId, idList.toArray());
                    }
                    return null;
                }

            });
        }
        return count;
    }

    @Override
    public List<AlertPrice> loadByUserId(String userId) {
        Map<Object, Object> map = this.redisTemplate.opsForHash().entries(CACHE_KEY_PREFIX + userId);
        if (map.isEmpty()) {
            List<AlertPrice> list = this.alertPriceMapper.loadByUserId(userId);
            this.cacheAlertPrice(true, true, list.toArray(new AlertPrice[] {}));
            return list;
        } else {
            List<AlertPrice> list = Lists.newArrayList();
            for (Map.Entry<Object, Object> entry : map.entrySet()) {
                Object value = entry.getValue();
                list.add(CopyBeanUtils.copy(value, AlertPrice.class));
            }
            return list;
        }
        // this.test();
        // this.putPair("BNBETH", 0.13345678);
        // this.queryAndFilterByLua("BNBBTC1", 0.00123, 0.4234987);
        // this.queryAndFilterByLua("BNBETH", 0.123456, 0.1234987);
        // return Lists.newArrayList();
    }

    @Override
    public void putPair(String pair, Double curPrice) {
        if (!alertPriceEnabled && (this.alertPriceAndroidEnabled || this.alertPriceIosEnabled
                || this.alertPriceWebEnabled || this.alertPricePcEnabled)) {
            throw new UnsupportedOperationException(
                    "jpush.alert.price.enabled must to be 'true' to enable the operation.");
        }
        log.info("插入收到的价格提醒交易对和最新价格到队列中：{},{}", pair, curPrice);
        ALERT_PRICE_PAIR_QUEUE.add(pair + ";" + decimal_10_8(curPrice));
    }

    @Override
    public void putAlertPrice(String alertPrice, Double curPrice) {
        if (!alertPriceEnabled && (this.alertPriceAndroidEnabled || this.alertPriceIosEnabled
                || this.alertPriceWebEnabled || this.alertPricePcEnabled)) {
            throw new UnsupportedOperationException(
                    "jpush.alert.price.enabled must to be 'true' to enable the operation.");
        }
        log.info("插入收到的待推送价格提醒信息到队列中：{},{}", alertPrice, curPrice);
        ALERT_PRICE_STRING_QUEUE.add(alertPrice + ";" + decimal_10_8(curPrice));
    }


    private void beforeSave(AlertPrice price, ProductItemVO product) {
        int decimalPlaces = product.getDecimalPlaces();
        price.setLastPrice(price.getLastPrice().setScale(decimalPlaces, BigDecimal.ROUND_HALF_UP));
        price.setPrice(price.getPrice().setScale(decimalPlaces, BigDecimal.ROUND_HALF_UP));
    }

    private void cacheAlertPrice(boolean cacheInstance, boolean setExpire, AlertPrice... price) {

        this.redisTemplate.execute(new SessionCallback<Void>() {

            @Override
            @SuppressWarnings({"rawtypes", "unchecked"})
            public Void execute(RedisOperations operations) throws DataAccessException {
                Map<String, List<AlertPrice>> map = Maps.newHashMap();
                Set<String> keys = null;
                if (setExpire) {
                    keys = Sets.newHashSet();
                }
                for (AlertPrice ap : price) {
                    String mapKey = CACHE_KEY_PREFIX + ap.getUserId();
                    if (cacheInstance) {
                        operations.opsForHash().put(mapKey, ap.getId(), ap);
                    }
                    if (setExpire) {
                        keys.add(mapKey);
                    }
                    String key = ap.getBaseAsset() + ap.getQuoteAsset() + ap.getSide();
                    List<AlertPrice> apList = map.get(key);
                    if (apList == null) {
                        apList = Lists.newLinkedList();
                        map.put(key, apList);
                    }
                    apList.add(ap);
                }
                if (setExpire) {
                    for (String key : keys) {
                        operations.expire(key, 48, TimeUnit.HOURS);
                    }
                }
                for (Iterator<String> iter = map.keySet().iterator(); iter.hasNext();) {
                    String key = iter.next();
                    Set<TypedTuple<Object>> tuples = Sets.newHashSet();
                    for (AlertPrice ap : map.get(key)) {
                        TypedTuple<Object> tuple = new DefaultTypedTuple<>(serializer(ap), ap.getPrice().doubleValue());
                        tuples.add(tuple);
                    }
                    operations.opsForZSet().add(key, tuples);
                }

                return null;
            }

        });
    }

    private void deleteByPrimaryKeyAndUserId(List<Long> ids, List<String> userIds) {
        this.redisTemplate.execute(new SessionCallback<Void>() {

            @SuppressWarnings({"rawtypes", "unchecked"})
            @Override
            public Void execute(RedisOperations operations) throws DataAccessException {
                for (int i = 0; i < ids.size(); i++) {
                    Long id = ids.get(i);
                    String userId = userIds.get(i);
                    operations.boundHashOps(CACHE_KEY_PREFIX + userId).delete(id);
                }
                return null;
            }

        });
        this.alertPriceMapper.deleteByPrimaryKeyAndUserId(ids, userIds);
    }

    // 从存放交易对和最新价格的队列中获取数据，然后计算，进而推送
    private void pushFromPairQueue() {
        while (this.alertPriceEnabled && (this.alertPriceAndroidEnabled || this.alertPriceIosEnabled
                || this.alertPriceWebEnabled || this.alertPricePcEnabled)) {
            try {
                String pairSting = null;
                try {
                    pairSting = ALERT_PRICE_PAIR_QUEUE.take();
                } catch (InterruptedException e) {
                    log.warn("等待从ALERT_PRICE_PAIR_QUEUE获取元素出现异常", e);
                    Thread.currentThread().interrupt();
                }
                if (pairSting == null) {
                    continue;
                }
                String[] array = StringUtils.split(pairSting, ";");
                String symbol = array[0];
                Double curPrice = Double.parseDouble(array[1]);
                String prevValue = (String) this.redisTemplate.opsForValue()
                        .getAndSet(LASTEST_SYMBOL_PRICE_KEY_PREV + "_" + symbol, String.valueOf(curPrice));
                List<Double> prevValues = Lists.newArrayList();
                if (StringUtils.isBlank(prevValue)) {
                    prevValues.add(0d);
                    prevValues.add(Double.MAX_VALUE);
                } else {
                    prevValues.add(Double.parseDouble(prevValue));
                }
                for (Double prevPrice : prevValues) {
                    if (Objects.equal(curPrice, prevPrice)) {
                        log.info("本次最新价格与最近一次价格相同，不进行推送：{},{}", symbol, curPrice);
                        continue;
                    }
                    // 计算要查询的价格区间==============================================
                    byte side = curPrice > prevPrice ? (byte) 1 : (byte) 2;
                    String key = symbol + side;
                    double min = side == 1 ? prevPrice : curPrice;
                    double max = side == 1 ? curPrice : prevPrice;
                    // 下面根据是否启用最大、最小值策略来设置本次查询的价格范围
                    if (this.minnumStrategy == 1) {
                        if (side == 1) {
                            // 上涨
                            min = 0;
                        } else {
                            // 下跌
                            min = 0;
                        }
                    }
                    if (this.maxnumStrategy == 1) {
                        if (side == 1) {
                            // 上涨
                            max = Double.MAX_VALUE;
                        } else {
                            // 下跌
                            max = curPrice;
                        }
                    }
                    // ===========================================================
                    Query query = new Query(key, min, max, /* threshold, count, */ curPrice);
                    query.start();
                }
            } catch (Exception e) {
                log.error("pushFromPairQueue()出现异常", e);
            }
        }

    }

    // 从存放待推送数据的队列中获取数据，然后计算，进而推送
    private void pushFromAlertPriceQueue() {
        PushBody body = new PushBody();
        while (this.alertPriceEnabled && (this.alertPriceAndroidEnabled || this.alertPriceIosEnabled
                || this.alertPriceWebEnabled || this.alertPricePcEnabled)) {
            try {
                String alertPriceString = null;
                if (body.size() == 0) {
                    try {
                        alertPriceString = ALERT_PRICE_STRING_QUEUE.take();
                    } catch (InterruptedException e) {
                        log.warn("等待从ALERT_PRICE_STRING_QUEUE获取元素出现异常", e);
                        Thread.currentThread().interrupt();
                    }
                } else {
                    try {
                        alertPriceString = ALERT_PRICE_STRING_QUEUE.poll(50, TimeUnit.MILLISECONDS);
                    } catch (InterruptedException e) {
                        log.warn("等待从ALERT_PRICE_STRING_QUEUE获取元素出现异常", e);
                        Thread.currentThread().interrupt();
                    }
                }
                if (alertPriceString == null) {
                    // 阻塞一定时间仍然没有获取到要推送的信息则将已经整理的消息推送出去
                    try {
                        this.push(body);
                    } finally {
                        body = new PushBody();
                    }
                    continue;
                } else {
                    String[] array = StringUtils.split(alertPriceString, ";");
                    Double curPrice = Double.parseDouble(array[1]);
                    body.addPayload(curPrice, array[0]);
                }
                if (this.batchCount > 0) {
                    // 支持批处理
                    if (body.size() == this.batchCount) {
                        try {
                            this.push(body);
                        } finally {
                            body = new PushBody();
                        }
                    }
                } else {
                    // 不支持批处理，则立即推送
                    try {
                        this.push(body);
                    } finally {
                        body = new PushBody();
                    }
                }
            } catch (Exception e) {
                body = new PushBody();
                log.error("pushFromAlertPriceQueue()出现异常", e);
            }
        }
    }

    private void success() {
        List<String> userIds = Lists.newLinkedList();
        List<Long> ids = Lists.newLinkedList();
        for (;;) {
            try {
                String alertPrice = null;
                if (userIds.isEmpty()) {
                    try {
                        alertPrice = ALERT_PRICE_SUCCESS_QUEUE.take();
                    } catch (Exception e) {
                        log.warn("等待从ALERT_PRICE_SUCCESS_QUEUE获取元素出现异常", e);
                        Thread.currentThread().interrupt();
                    }
                } else {
                    try {
                        alertPrice = ALERT_PRICE_SUCCESS_QUEUE.poll(100, TimeUnit.MILLISECONDS);
                    } catch (InterruptedException e) {
                        log.warn("等待从ALERT_PRICE_SUCCESS_QUEUE获取元素出现异常", e);
                        Thread.currentThread().interrupt();
                    }
                }
                if (alertPrice == null) {
                    if (userIds.size() > 0) {
                        try {
                            this.deleteByPrimaryKeyAndUserId(ids, userIds);
                        } finally {
                            userIds.clear();
                            ids.clear();
                        }
                    }
                } else {
                    AlertPrice ap = deserializer(alertPrice);
                    userIds.add(ap.getUserId());
                    ids.add(ap.getId());
                }
                if (userIds.size() == 500) {
                    try {
                        this.deleteByPrimaryKeyAndUserId(ids, userIds);
                    } finally {
                        userIds.clear();
                        ids.clear();
                    }
                }
            } catch (Exception e) {
                log.error("success()出现异常", e);
            }
        }
    }


    private void failure() {
        List<AlertPrice> apList = Lists.newLinkedList();
        for (;;) {
            try {
                String alertPrice = null;
                if (apList.isEmpty()) {
                    try {
                        alertPrice = ALERT_PRICE_FAILURE_QUEUE.take();
                    } catch (InterruptedException e) {
                        log.warn("等待从ALERT_PRICE_FAILURE_QUEUE获取元素出现异常", e);
                        Thread.currentThread().interrupt();
                    }
                } else {
                    try {
                        alertPrice = ALERT_PRICE_FAILURE_QUEUE.poll(100, TimeUnit.MILLISECONDS);
                    } catch (InterruptedException e) {
                        log.warn("等待从ALERT_PRICE_FAILURE_QUEUE获取元素出现异常", e);
                        Thread.currentThread().interrupt();
                    }
                }
                if (alertPrice == null) {
                    if (apList.size() > 0) {
                        try {
                            this.cacheAlertPrice(false, false, apList.toArray(new AlertPrice[] {}));
                        } finally {
                            apList.clear();
                        }
                    }
                } else {
                    AlertPrice ap = deserializer(alertPrice);
                    apList.add(ap);
                }
                if (apList.size() == 5000) {
                    try {
                        this.cacheAlertPrice(false, false, apList.toArray(new AlertPrice[] {}));
                    } finally {
                        apList.clear();
                    }
                }
            } catch (Exception e) {
                log.error("failure()出现异常", e);
            }
        }
    }

    private Collection<String> queryAndFilterByLua(String key, double min, double max) {
        log.info("开始根据价格范围查询符合推送条件的数据：{},{},{}", key, min, max);
        StopWatch watch = new StopWatch();
        List<String> list = this.redisTemplate.execute(new RedisCallback<List<String>>() {

            @Override
            public List<String> doInRedis(RedisConnection connection) throws DataAccessException {
                Object result = null;
                try {
                    watch.start("==========开始通过默认RedisConnection执行lua脚本查询");
                    try {
                        result = connection.eval(luaScript.getBytes(), ReturnType.fromJavaType(List.class), 1,
                                key.getBytes(), decimal_10_8(min).getBytes(), decimal_10_8(max).getBytes());
                    } finally {
                        watch.stop();
                    }
                } catch (InvalidDataAccessApiUsageException | UnsupportedOperationException e) {
                    Object nativeConn = connection.getNativeConnection();
                    if (nativeConn instanceof JedisCluster) {
                        JedisCluster cluster = (JedisCluster) nativeConn;
                        try {
                            watch.start("==========开始通过JedisCluster执行lua脚本查询");
                            try {
                                result = cluster.eval(luaScript, Arrays.asList(key),
                                        Arrays.asList(decimal_10_8(min), decimal_10_8(max)));
                            } finally {
                                watch.stop();
                            }
                        } catch (InvalidDataAccessApiUsageException | UnsupportedOperationException e2) {
                            watch.start("==========开始通过直接redis命令查询");
                            try {
                                List<String> list = queryAndFilter(key, min, max);
                                return list;
                            } finally {
                                watch.stop();
                            }
                        }
                    }
                }
                if (result != null) {
                    return deserializeResult(result);
                }
                return null;
            }

        });
        log.info("根据条件查询符合推送条件的数据完成：{},{},{},{}条，执行耗时{}", key, min, max, list.size(), watch.prettyPrint());
        return list;
    }

    private List<String> queryAndFilter(String key, double min, double max) {
        return this.redisTemplate.execute(new SessionCallback<List<String>>() {

            @SuppressWarnings({"unchecked", "rawtypes"})
            @Override
            public List<String> execute(RedisOperations operations) throws DataAccessException {
                Object value = operations.boundZSetOps(key).rangeByScore(min, max);
                operations.boundZSetOps(key).removeRangeByScore(min, max);
                return deserializeResult(value);
            }

        });
    }

    private List<String> deserializeResult(Object result) {
        List<String> results = Lists.newLinkedList();
        for (Object obj : (List<?>) result) {
            String item = obj.toString();
            item = StringUtils.remove(item, "\"");
            results.add(item);
        }
        return results;
    }

    private void doTrigger(Collection<String> cols, Double curPrice) {
        log.info("查询到符合推送条件的数据有{}条，当前价格是：", cols.size(), curPrice);
        int i = 0;
        for (String str : cols) {
            if (this.alertByKafka) {// 支持kafka广播待推送消息
                // 每隔2条数据，将待处理消息放到本服务节点的队列中，其他的待推送消息通过kafka广播
                if (i++ % 3 == 0) {
                    this.putAlertPrice(str, curPrice);
                } else {
                    this.sendToKafka(curPrice, str);
                }
            } else {// 不支持kafka广播待推送消息，直接放到本服务节点的队列中
                this.putAlertPrice(str, curPrice);
            }
        }
    }

    private void sendToKafka(Double curPrice, String alertPrice) {
        this.productor.sendAlertPriceMessage(alertPrice + ";" + decimal_8(curPrice));
    }

    private String serializer(AlertPrice ap) {
        String price = decimal_10_8(ap.getPrice());
        String lastPrice = decimal_10_8(ap.getLastPrice());
        return ap.getStatus() + "|" + ap.getSide() + "|" + price + "|" + lastPrice + "|" + ap.getUserId() + "|"
                + ap.getId() + "|" + ap.getBaseAsset() + "|" + ap.getQuoteAsset();
    }

    private AlertPrice deserializer(String apStr) {
        String[] array = StringUtils.split(apStr, "|");
        AlertPrice ap = new AlertPrice();
        ap.setStatus(Byte.parseByte(array[0]));
        ap.setSide(Byte.parseByte(array[1]));
        ap.setPrice(BigDecimal.valueOf(Double.parseDouble(array[2])));
        ap.setLastPrice(BigDecimal.valueOf(Double.parseDouble(array[3])));
        ap.setUserId(array[4]);
        ap.setId(Long.parseLong(array[5]));
        ap.setBaseAsset(array[6]);
        ap.setQuoteAsset(array[7]);
        return ap;
    }

    public void test() {
        if (!StringUtils.equals(this.env.getProperty("active"), "prod")) {
            for (int i = 0; i < 10; i++) {
                AlertPrice ap = new AlertPrice();
                ap.setId(Long.valueOf(i));
                ap.setBaseAsset("BNB");
                ap.setQuoteAsset("ETH");
                ap.setSide((byte) (i % 2 == 0 ? 1 : 2));
                ap.setUserId("35000132");
                ap.setPrice(BigDecimal.valueOf(i + 0.0001));
                ap.setLastPrice(ap.getSide() == 1 ? BigDecimal.valueOf(i) : BigDecimal.valueOf(i + 0.1));

                // this.putAlertPrice(serializer(ap), 0.01234567);
                // this.sendToKafka(0.01234567, serializer(ap));
                try {
                    this.insert(ap);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }

    private MultipleNotificationPacket createPacket(Double curPrice, AlertPrice ap) {
        if (!(this.alertPriceEnabled && (this.alertPriceAndroidEnabled || this.alertPriceIosEnabled
                || this.alertPricePcEnabled || this.alertPriceWebEnabled)) || StringUtils.isBlank(ap.getUserId())) {
            return null;
        }
        String title = "#{pa.title}";
        String content = null;
        String symbol = ap.getBaseAsset() + ap.getQuoteAsset();
        if (ap.getSide() == 1) {
            content = "#{pa.up:" + ap.getBaseAsset() + "," + ap.getQuoteAsset() + "," + ap.getPrice()
                    .stripTrailingZeros().toPlainString() /*
                                                           * + "," + this.productRealtimePriceService.getUsd(symbol)
                                                           */ + "}";
        } else {
            content = "#{pa.down:" + ap.getBaseAsset() + "," + ap.getQuoteAsset() + "," + ap.getPrice()
                    .stripTrailingZeros().toPlainString() /*
                                                           * + "," + this.productRealtimePriceService.getUsd(symbol)
                                                           */ + "}";
        }
        PushCommand command = new PushCommand();
        command.getUsers().add(ap.getUserId());
        command.setTagNotList(Arrays.asList("PA"));
        MultipleNotificationPacket packet = new MultipleNotificationPacket();
        packet.setCommand(command);
        packet.setSendno(ServiceHelper.generateSendno());
        if (this.alertPriceAndroidEnabled && this.alertPriceIosEnabled) {
            AppPayload appMessage = new AppPayload();
            appMessage.setTitle(title);
            appMessage.setContent(content);
            packet.setAppMessage(appMessage);
        } else {
            AndroidPayload androidMessage = null;
            if (this.alertPriceAndroidEnabled) {
                androidMessage = new AndroidPayload();
                androidMessage.setTitle(title);
                androidMessage.setContent(content);
                packet.setAndroidMessage(androidMessage);
            }
            IosPayload iosMessage = null;
            if (this.alertPriceIosEnabled) {
                iosMessage = new IosPayload();
                iosMessage.setTitle(title);
                iosMessage.setContent(content);
                packet.setIosMessage(iosMessage);
            }
        }
        PcPayload pcMessage = null;
        if (this.alertPricePcEnabled) {
            pcMessage = new PcPayload();
            pcMessage.setTitle(title);
            pcMessage.setContent(content);
            packet.setPcMessage(pcMessage);
        }
        WebPayload webMessage = null;
        if (this.alertPriceWebEnabled) {
            webMessage = new WebPayload();
            webMessage.setTitle(title);
            webMessage.setContent(content);
            packet.setWebMessage(webMessage);
        }
        return packet;
    }

    private void push(PushBody body) {
        log.info("即将调用推送服务推送价格提醒信息，本批次有{}条", body.size());
        if (body.size() > 0) {
            log.info("调用发送服务发送数据：{}条", body.size());
            this.pushService.sendMultiple(body.pushPayload, new RequestCallback() {

                @Override
                public void completed(List<PushResult> resultList, Object... obj) {
                    log.info("调用发送服务发送数据返回：{}条结果", resultList.size());
                    onPush(resultList, body);
                }

            });
        }
    }

    private void onPush(List<PushResult> resultList, PushBody body) {
        for (PushResult pushResult : resultList) {
            if (!pushResult.isResultOK()) {
                if (pushResult.getEx() != null) {
                    log.error("发送出现异常：{}，{}", pushResult, pushResult.getEx());
                }
            }
            // 只有推送失败的才会返回具体的结果。需要将这些alertPrice信息重新存入redis中
            for (Iterator<PushResult> iter = pushResult.errIterator(); iter.hasNext();) {
                PushResult result = iter.next();
                AlertPrice alertPrice = body.alertPriceMap.remove(result.sendno);
                if (alertPrice != null) {
                    ALERT_PRICE_FAILURE_QUEUE.add(serializer(alertPrice));
                }
            }
        }
        // body.alertPriceMap中剩下的都是成功推送的，需要从数据库中删除
        for (Iterator<Map.Entry<Integer, AlertPrice>> iter = body.alertPriceMap.entrySet().iterator(); iter
                .hasNext();) {
            Map.Entry<Integer, AlertPrice> entry = iter.next();
            AlertPrice alertPrice = entry.getValue();
            if (alertPrice != null) {
                ALERT_PRICE_SUCCESS_QUEUE.add(serializer(alertPrice));
            }
            iter.remove();
        }
    }

    private class Query {

        volatile boolean stop;
        // int threadCount;
        String key;
        double min;
        double max;
        // long threshold;
        double curPrice;

        Query(String key, double min, double max, /* long threshold, int threadCount, */double curPrice) {
            this.key = key;
            this.max = max;
            this.min = min;
            // this.threadCount = threadCount > 8 ? 8 : threadCount;
            // this.threshold = threshold;
            this.curPrice = curPrice;
        }

        public void start() {
            // for (int i = 0; i < this.threadCount && !stop; i++) {
            // Worker.submit(new Runnable() {
            // @Override
            // public void run() {
            // while (!stop && !Thread.interrupted()) {
            // execute();
            // }
            // }
            // });
            // }
            this.execute();
        }

        private void execute() {
            Collection<String> cols = null;
            try {
                cols = queryAndFilterByLua(key, min, max/* , 0, threshold */);
            } catch (Exception e) {
                log.error("从redis中加载待推送价格提醒信息出错", e);
                stop = true;
                return;
            }
            if (cols == null || cols.isEmpty() /* || cols.size() < threshold */) {
                // 如果查询结果数据量小于阈值，则表示没有更多的数据了，因此尚未执行的线程就不再处理实际的查询操作
                stop = true;
            }
            if (cols != null) {
                try {
                    doTrigger(cols, curPrice);
                } catch (Exception e) {
                    log.error("调用doTrigger方法出错", e);
                }
            }
        }
    }

    private class PushBody {
        List<MultipleNotificationPacket> pushPayload = Lists.newLinkedList();
        Map<Integer, AlertPrice> alertPriceMap = Maps.newHashMap();

        public void addPayload(Double curPrice, String alertPrice) {
            AlertPrice ap = deserializer(alertPrice);
            MultipleNotificationPacket packet = createPacket(curPrice, ap);
            if (packet != null) {
                this.alertPriceMap.put(packet.getSendno(), ap);
                this.pushPayload.add(packet);
            }
        }

        public int size() {
            return this.alertPriceMap.size();
        }
    }

    @Override
    public List<AlertPrice> query(AlertPrice alertPrice) {
        // TODO Auto-generated method stub
        List<AlertPrice> alertPriceList = alertPriceMapper.query(alertPrice);
        return alertPriceList;
    }

    @Override
    public Long count(AlertPrice alertPrice) {
        // TODO Auto-generated method stub
        return alertPriceMapper.count(alertPrice);
    }

}
